package bigdata.scala.spark.streaming.ycfx.userstatusqueue

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util

import bigdata.java.framework.spark.kafka.{SKProcessor, SimpleSparkKafka}
import bigdata.java.framework.spark.util.client.{HBaseClient, KafkaProducerClient}
import bigdata.scala.spark.streaming.ycfx.model.{UserPowerCurveModel, UserPowerModel, UserStatus}
import com.alibaba.fastjson.JSON
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.api.java.JavaInputDStream
import org.apache.spark.streaming.dstream.DStream
import bigdata.scala.spark.streaming.ycfx.model.MyConstants

/**
 * @author : lubohua
 * @date : 2020/1/17 9:17
 * @description : 异常电压处理
 * 参数示例 : appname: UserStatus topics:Data1.1 groupId: duration: maxnum:
 */
object UserPowerQueue {
  def main(args: Array[String]): Unit = {
    //从kafka获取用户电压数据
    val simpleSparkKafka = new SimpleSparkKafka(args)
    simpleSparkKafka.execute(new SKProcessor {
      override def process(directStream: JavaInputDStream[ConsumerRecord[String, String]]): Unit = {
        val stream = directStream.inputDStream
        val baseStream: DStream[String] = stream.map(cr=>cr.value())
        dataShareQueue(baseStream)
      }
    }).start()
  }

  //数据转换 将从kafka获取到的json字符串转为string类型
  def dataShareQueue(baseStream: DStream[String]): Unit ={
    baseStream.foreachRDD(rdd=>{
      rdd.map(kafka => JSON.parseObject(kafka))
        .map(json => {
          val meterId = json.getString("METER_ID")
          val meterAssetno = json.getString("METER_ASSETNO")
          val dataDate = json.getString("DATA_DATE")
          val phase = json.getString("PHASE")
          val dotdensity = json.getString("DOTDENSITY")
          val u1 = json.getString("U1")
          val u5 = json.getString("U5")
          val u9 = json.getString("U9")
          val u13 = json.getString("U13")
          val u17 = json.getString("U17")
          val u21 = json.getString("U21")
          val u25 = json.getString("U25")
          val u29 = json.getString("U29")
          val u33 = json.getString("U33")
          val u37 = json.getString("U37")
          val u41 = json.getString("U41")
          val u45 = json.getString("U45")
          val u49 = json.getString("U49")
          val u53 = json.getString("U53")
          val u57 = json.getString("U57")
          val u61 = json.getString("U61")
          val u65 = json.getString("U65")
          val u69 = json.getString("U69")
          val u73 = json.getString("U73")
          val u77 = json.getString("U77")
          val u81 = json.getString("U81")
          val u85 = json.getString("U85")
          val u89 = json.getString("U89")
          val u93 = json.getString("U93")
          val userPowerCurveModel = UserPowerCurveModel(meterId, meterAssetno, dataDate, dotdensity, phase, u1, u5, u9, u13, u17, u21, u25, u29, u33, u37, u41, u45, u49, u53, u57, u61, u65, u69, u73, u77, u81, u85, u89, u93)
          putUserPowerHbase(userPowerCurveModel)
        })
    })
  }

  // 将用户电压写入hbase
  def putUserPowerHbase(userPowerCurveModel: UserPowerCurveModel) = {
    val dataDate:String = LocalDateTime.parse(userPowerCurveModel.dataDate, DateTimeFormatter.ofPattern("yyyyMMdd")).toString // 获取时间
    putUserPowerHbaseBy24Point(dataDate, userPowerCurveModel)
  }

  // 将24个点用户电压写入hbase
  def putUserPowerHbaseBy24Point(dataDate : String, userPowerCurveModel : UserPowerCurveModel): Unit = {
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 1, userPowerCurveModel.u1, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第一个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 5, userPowerCurveModel.u5, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第五个点到hbase
    // 其他同理
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 9, userPowerCurveModel.u9, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第九个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 13, userPowerCurveModel.u13, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第十三个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 17, userPowerCurveModel.u17, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第十七个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 21, userPowerCurveModel.u21, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第二十一个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 25, userPowerCurveModel.u25, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第二十五个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 29, userPowerCurveModel.u29, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第二十九个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 33, userPowerCurveModel.u33, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第三十三个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 37, userPowerCurveModel.u37, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第三十七个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 41, userPowerCurveModel.u41, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第四十一个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 45, userPowerCurveModel.u45, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第四十五个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 49, userPowerCurveModel.u49, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第四十九个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 53, userPowerCurveModel.u53, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第五十三个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 57, userPowerCurveModel.u57, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第五十七个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 61, userPowerCurveModel.u61, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第六十一个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 65, userPowerCurveModel.u65, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第六十五个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 69, userPowerCurveModel.u69, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第六十九个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 73, userPowerCurveModel.u73, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第七十三个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 77, userPowerCurveModel.u77, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第七十七个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 81, userPowerCurveModel.u81, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第八十一个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 85, userPowerCurveModel.u85, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第八十五个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 89, userPowerCurveModel.u89, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第八十九个点到hbase
    putUserPowerHbaseByPoint(dataDate, userPowerCurveModel.phase, 93, userPowerCurveModel.u93, userPowerCurveModel.meterId, userPowerCurveModel.meterAssetno) // 写入第九十三个点到hbase
  }

  // 将一个点的数据写入hbase
  def putUserPowerHbaseByPoint(dataDate : String, phase : String, point : Int, powerValue : String, meterId : String, meterAssetno : String): Unit = {
    if(!(powerValue==null||powerValue.equals(""))){
      // 通过点数生成rowkey
      val rowkey = genRowkey(dataDate, point)
      //列族
      val columnFamily = ""
      // 从hbase获取用户电压
      val userPower :String= getUserPowerHbase(rowkey)
      // 生成用户电压数据
      val model = genUserPower(userPower, phase,point,powerValue, meterId, meterAssetno)
      //将基础电压数据放入hbase
      putUserPowerHbase(MyConstants.tab_user_power,rowkey,columnFamily,model)

      //判断电压是否异常
      if (getAbnormalUserPower(model)){
        //说明电压异常  将异常电压数据写入hbase异常电压表
        putExUserPowerHbase(MyConstants.tab_user_power_abnormal,rowkey,columnFamily,model)
        putExUserPowerKafka(model)
      }


    }
  }

  //将基础电压数据写入hbase
  def putUserPowerHbase(tableName:String,rowKey:String,columnFamily:String,model: UserPowerModel): Unit ={
    val userPower: String = JSON.toJSONString(model)
//    val userPowerMap:util.HashMap[String,String] = new util.HashMap[String,String]()
//    userPowerMap.put("meterId",model.meterId)
//    userPowerMap.put("meterAssetno",model.meterAssetno)
//    userPowerMap.put("aphase",model.aphase)
//    userPowerMap.put("bphase",model.bphase)
//    userPowerMap.put("cphase",model.cphase)
//    HBaseClient.getInstance().put(tableName,rowKey,columnFamily,userPowerMap)
    HBaseClient.getInstance().putCfDefault(tableName,rowKey,userPower)
  }

  //将异常电压数据写入hbase
  def putExUserPowerHbase(tableName:String,rowKey:String,columnFamily:String,model: UserPowerModel): Unit ={
    val exUserPower = JSON.toJSONString(model)
//    val userPowerMap:util.HashMap[String,String] = new util.HashMap[String,String]()
//    userPowerMap.put("meterId",model.meterId)
//    userPowerMap.put("meterAssetno",model.meterAssetno)
//    userPowerMap.put("aphase",model.aphase)
//    userPowerMap.put("bphase",model.bphase)
//    userPowerMap.put("cphase",model.cphase)
//    HBaseClient.getInstance().put(tableName,rowKey,columnFamily,userPowerMap)
    HBaseClient.getInstance().putCfDefault(tableName,rowKey,exUserPower)
  }

  //将异常电压数据下发通知至kafka
  def putExUserPowerKafka(model: UserPowerModel): Unit ={
    val userPowerMap:util.HashMap[AnyRef,AnyRef] = new util.HashMap[AnyRef,AnyRef]()
    userPowerMap.put("meterId",model.meterId)
    userPowerMap.put("meterAssetno",model.meterAssetno)
    userPowerMap.put("aphase",model.aphase)
    userPowerMap.put("bphase",model.bphase)
    userPowerMap.put("cphase",model.cphase)
    KafkaProducerClient.getInstance().sendMessage(MyConstants.tpc_user_power_abnormal,userPowerMap)
  }

  // 生成rowkey,point代表是第几个点 u1是第一个点 以此类推
  def genRowkey(dataDate : String, point : Int) = {
    val hourNumber = (point - 1) * 15 / 60; //十五分钟一个点，求出来的就是第几个小时
    //var dataDateTime = dataDate + (hourNumber < 10 ? "0" + hourNumber.toString : hourNumber.toString)
    val dataDateTime = dataDate + (if (hourNumber<10) "0"+hourNumber.toString else hourNumber.toString)
    dataDateTime
  }

  // 根据rowkey获取hbase中用户电压
  def getUserPowerHbase(rowkey : String):String = {
    var userPower:String = null
    //如果hbase有则取hbase中的数据
    if (HBaseClient.getInstance().exists(MyConstants.tab_user_power, rowkey)) {
      userPower = HBaseClient.getInstance().getCfDefault(MyConstants.tab_user_power,rowkey)
    }
    userPower
  }

  // 生成电压数据
  def genUserPower(userPower :String , phase : String, point : Int, powerValue : String, meterId : String, meterAssetno : String): UserPowerModel = {
    // 如果不存在用户电压则初始化用户电压
    var userPowerModel:UserPowerModel = null
    if (userPower == null) {
      userPowerModel.meterId = meterId
      userPowerModel.meterAssetno = meterAssetno
      phase match {
        case "1" => {
          userPowerModel.aphase = powerValue
        }
        case "2" => {
          userPowerModel.bphase = powerValue
        }
        case "3" => {
          userPowerModel.cphase = powerValue
        }
      }
    }

    userPowerModel
  }

  //判断电压是否异常 电压>235<198则为异常
  def getAbnormalUserPower(userPowerModel: UserPowerModel): Boolean ={
    val boolean:Boolean = true
    val aphase = userPowerModel.aphase.toDouble
    val bphase = userPowerModel.bphase.toDouble
    val cphase = userPowerModel.cphase.toDouble
    if (aphase>235||aphase<198||aphase==null){
      boolean == false
    }
    if (bphase>235||bphase<198||bphase==null){
      boolean == false
    }
    if (cphase>235||bphase<198||cphase==null){
      boolean == false
    }
    boolean
  }



}
